import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import com.alibaba.fastjson.JSON;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

public class bolt1_2 extends BaseRichBolt {
    private static final long serialVersionUID = 886149197481637894L;
    private OutputCollector collector;
    //设置hashmap用于接收书籍以及出版社（销量）数据同时进行排名
    private Map<String, Integer> book_sum = new HashMap<String, Integer>();
    private Map<String, Integer> press_sum = new HashMap<String, Integer>();
    private Map<String, Integer> type_sum = new HashMap<String, Integer>();
    private Map<String, AtomicInteger> counterMap;

    public void prepare(Map stormConf, TopologyContext context,
                        OutputCollector collector) {
        this.collector = collector;
        this.counterMap = new HashMap<String, AtomicInteger>();
    }

    public void execute(Tuple input) {
        try{
            //设置变量接收处理完的数据，例如：top10书籍及其销量
            String list1 = null;
            String list2 = null;
            String list3 = null;
            String list1_1 = null;
            String list2_1 = null;
            String list3_1 = null;
            //接收上一个bolt处理完后的数据
            String book_num_1 = input.getStringByField("book_num");
            String book_type = input.getStringByField("book_type");
            String book_name=input.getStringByField("book_name").replaceAll("^\"|\"$", "");
            String book_press=input.getStringByField("book_press").replaceAll("^\"|\"$", "");
            //String转int
            int book_num=Integer.parseInt(book_num_1);
            int num=0;
            if (book_sum.containsKey(book_name)) {
                num = book_sum.get(book_name);
            }
            num+=book_num;
            book_sum.put(book_name,num);
            //取出销量前十的书籍
            String regEx = "[\n`\"\"/”“\\\\\\\\]";
            try {
                //hash map按key进行从大到小的排序
                List<String> mobileList = book_sum.entrySet().stream()
                        .sorted((Map.Entry<String, Integer> e1, Map.Entry<String, Integer> e2) -> e2.getValue() - e1.getValue())
                        .map(entry -> entry.getKey()).collect(Collectors.toList())
                        .subList(0, 10);
                List<Integer> mobileList_1 = book_sum.entrySet().stream()
                        .sorted((Map.Entry<String, Integer> e1, Map.Entry<String, Integer> e2) -> e2.getValue() - e1.getValue())
                        .map(entry -> entry.getValue()).collect(Collectors.toList())
                        .subList(0, 10);
                list1_1=JSON.toJSONString(mobileList_1);
                list1=JSON.toJSONString(mobileList);
                list1 = list1.replaceAll(regEx, "");
            }catch (Exception e){}
            //统计前6的销售类型
            int num_3=0;
            if (type_sum.containsKey(book_type)) {
                num_3 = type_sum.get(book_type);
            }
            num_3+=book_num;
            type_sum.put(book_type,num_3);
            try {
                //hash map按key进行从大到小的排序
                List<String> mobileList = type_sum.entrySet().stream()
                        .sorted((Map.Entry<String, Integer> e1, Map.Entry<String, Integer> e2) -> e2.getValue() - e1.getValue())
                        .map(entry -> entry.getKey()).collect(Collectors.toList())
                        .subList(0, 6);
                List<Integer> mobileList_1 = type_sum.entrySet().stream()
                        .sorted((Map.Entry<String, Integer> e1, Map.Entry<String, Integer> e2) -> e2.getValue() - e1.getValue())
                        .map(entry -> entry.getValue()).collect(Collectors.toList())
                        .subList(0, 6);
                //System.out.println(mobileList_1);
                list3_1=JSON.toJSONString(mobileList_1);
                //System.out.println(list3_1);
                list3=JSON.toJSONString(mobileList);
                list3 = list3.replaceAll(regEx, "");
                //System.out.println(list3);
            }catch (Exception e){}
            //计算出版商的销售数量
            int num_2=0;
            if (press_sum.containsKey(book_press)) {
                num_2 = press_sum.get(book_press);
            }
            num_2+=book_num;
            press_sum.put(book_press,num_2);
            //取出销量前十的出版商
            try {
                //hash map按key进行从大到小的排序
                List<String> mobileList_2 = press_sum.entrySet().stream()
                        .sorted((Map.Entry<String, Integer> e1, Map.Entry<String, Integer> e2) -> e2.getValue() - e1.getValue())
                        .map(entry -> entry.getKey()).collect(Collectors.toList())
                        .subList(0,10);
                List<Integer> mobileList_1 = press_sum.entrySet().stream()
                        .sorted((Map.Entry<String, Integer> e1, Map.Entry<String, Integer> e2) -> e2.getValue() - e1.getValue())
                        .map(entry -> entry.getValue()).collect(Collectors.toList())
                        .subList(0, 10);
                list2_1=JSON.toJSONString(mobileList_1);
                list2 = JSON.toJSONString(mobileList_2);
                list2 = list2.replaceAll(regEx, "");
                //将结果写入数据库中
                try{
                    //以映射方式加载Driver,抛出异常
                    Class.forName("com.mysql.jdbc.Driver");
                    //建立连接
                    Connection conn = DriverManager.getConnection("jdbc:mysql://slave1:3306/storm1?useUnicode=true&characterEncoding=gbk", "root", "root");
                    //执行sql语句
                    String sql="insert into order_1 (top_book,top_press,top_type,book_num,press_num,type_num) values ("+"\""+list1+"\""+",\""+list2+"\""+",\""+list3+"\""+",\""+list1_1+"\""+",\""+list2_1+"\""+",\""+list3_1+"\");";
                    //通过创建的连接对象创建执行器,用来执行sql语句\
                    //System.out.println(sql);
                    Statement statement = conn.createStatement();
                    statement.executeUpdate(sql);
                    //用于调试输出sql语句
                    System.out.println(sql);
                    conn.close();
                    statement.close();}
                catch (Exception e){}
            }catch (Exception e){}
            //成功操作后返回ack信号
            collector.ack(input);}
        catch (Exception e){
            //操作失败后返回fail信号
            collector.fail(input);
        }
    }

    public void cleanup() {
        Iterator<Entry<String, AtomicInteger>> iter = this.counterMap
                .entrySet().iterator();
        while (iter.hasNext()) {
            Entry<String, AtomicInteger> entry = iter.next();
        }
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    }
}